package com.mrs.gauss.controller;

import com.mrs.gauss.common.BaseResponse;
import com.mrs.gauss.common.ErrorCode;
import com.mrs.gauss.common.ResultUtils;
import com.mrs.gauss.exception.BusinessException;
import com.mrs.gauss.interfaces.rest.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.servlet.http.HttpServletRequest;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;


/**
 * @author wubai
 */
@Slf4j
@RestController
//@RequestMapping(value = "/api/mrs", consumes = {"application/x-www-form-urlencoded"},produces = {"application/json" }, method = RequestMethod.POST)
@RequestMapping("/api/mrs")
public class MrsMasterController {
    private static Boolean isFirstgetHiveAuditList = true;
    AtomicReference<List<Map>> accessTables = new AtomicReference<>(new ArrayList<>());
    private static Boolean isFirstgetServerAlarmsList = true;
    AtomicReference<List<Map>> serverStatus = new AtomicReference<>(new ArrayList<>());
    private static Boolean isFirstgetNodesAlarmsList = true;
    AtomicReference<List<Map>> nodesStatus = new AtomicReference<>(new ArrayList<>());
    private static Boolean isFirstgetInstancesAlarmsList = true;
    AtomicReference<List<Map>> instancesStatus = new AtomicReference<>(new ArrayList<>());
    private static Boolean isFirstgetHdfsSummary = true;
    AtomicReference<List<Map>> hdfsSummary = new AtomicReference<>(new ArrayList<>());

    /**
     * 服务状态接口
     */
    @GetMapping("/servers")
    public BaseResponse<List<Map>> getServerAlarmsList(HttpServletRequest request) {

        log.info("=>调用: /getServerAlarmsList");
        if (isFirstgetServerAlarmsList) {
            isFirstgetServerAlarmsList = false;
            //  创建定时器
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdown));

            scheduler.scheduleAtFixedRate(() -> {
                try {
                    serverStatus.set(new MrsClusterServerAlarms().getServerAlarmsList());
                } catch (Exception e) {
                    throw new BusinessException(ErrorCode.PARAMS_ERROR, "getServerAlarmsList failed");
                }
            }, 0, 3, TimeUnit.MINUTES);
            return ResultUtils.success(serverStatus.get());
        } else {
            return ResultUtils.success(serverStatus.get());
        }

    }

    /**
     * 节点状态接口
     */
    @GetMapping("/nodes")
    public BaseResponse<List<Map>> getNodesAlarmsList(HttpServletRequest request) {

        log.info("=>调用: /getNodesAlarmsList");

        if (isFirstgetNodesAlarmsList) {
            isFirstgetNodesAlarmsList = false;
            //  创建定时器
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdown));

            scheduler.scheduleAtFixedRate(() -> {
                try {
                    nodesStatus.set(new MrsClusterNodesAlarms().getNodesAlarmsList());
                } catch (Exception e) {
                    throw new BusinessException(ErrorCode.PARAMS_ERROR, "getNodesAlarmsList failed");
                }
            }, 0, 3, TimeUnit.MINUTES);
            return ResultUtils.success(nodesStatus.get());
        } else {
            return ResultUtils.success(nodesStatus.get());
        }
    }

    /**
     * 实例状态接口
     */
    @GetMapping("/instances")
    public BaseResponse<List<Map>> getInstancesAlarmsList(HttpServletRequest request) {

        log.info("=>调用: /getInstancesAlarmsList");
        if (isFirstgetInstancesAlarmsList) {
            isFirstgetInstancesAlarmsList = false;
            //  创建定时器
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdown));

            scheduler.scheduleAtFixedRate(() -> {
                try {
                    instancesStatus.set(new MrsClusterInstancesAlarms().getInstancesAlarmsList());
                } catch (Exception e) {
                    throw new BusinessException(ErrorCode.PARAMS_ERROR, "getInstancesAlarmsList failed");
                }
            }, 0, 3, TimeUnit.MINUTES);
            return ResultUtils.success(instancesStatus.get());
        } else {
            return ResultUtils.success(instancesStatus.get());
        }
    }

    /**
     * hive访问次数接口
     */
    @GetMapping("/hive")
    public BaseResponse<List<Map>> getHiveAuditList(HttpServletRequest request) {

        log.info("=>调用: /getHiveAuditList");

        if (isFirstgetHiveAuditList) {
            isFirstgetHiveAuditList = false;
            //  创建定时器
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdown));

            scheduler.scheduleAtFixedRate(() -> {
                try {
                    accessTables.set(new AuditHiveApi().getAccessTableList());
                } catch (Exception e) {
                    throw new BusinessException(ErrorCode.PARAMS_ERROR, "getHiveAuditList failed");
                }
            }, 0, 3, TimeUnit.MINUTES);
            return ResultUtils.success(accessTables.get()); //to do 返回值类型是否List<Map>
        } else {
            return ResultUtils.success(accessTables.get());
        }
    }
    /**
     * hdfs-summary接口
     */
    @GetMapping("/summary")
    public BaseResponse<List<Map>> getHdfsSummary(HttpServletRequest request) {

        log.info("=>调用: /getHdfsSummary");

        if (isFirstgetHdfsSummary) {
            isFirstgetHdfsSummary = false;
            //  创建定时器
            ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
            Runtime.getRuntime().addShutdownHook(new Thread(scheduler::shutdown));

            scheduler.scheduleAtFixedRate(() -> {
                try {
                    hdfsSummary.set(new HdfsSummary().getHdfsSummary());
                } catch (Exception e) {
                    throw new BusinessException(ErrorCode.PARAMS_ERROR, "getHdfsSummary failed");
                }
            }, 0, 3, TimeUnit.MINUTES);
            return ResultUtils.success(hdfsSummary.get()); //to do 返回值类型是否List<Map>
        } else {
            return ResultUtils.success(hdfsSummary.get());
        }
    }
    /**
     * yarn接口
     */
    @GetMapping("/yarn")
    public BaseResponse<List<Map>> getYarnRunningTaskList(HttpServletRequest request)  {
        log.info("=>调用: /getYarnRunningTaskList");
        try {
            YarnRunning yarnRunning = new YarnRunning();
            List<Map> yarnRunningTaskList = yarnRunning.getYarnRunningTaskList();
            if (yarnRunningTaskList == null) {
                throw new BusinessException(ErrorCode.PARAMS_ERROR, "Yarn running task list is null");
            }
            return ResultUtils.success(yarnRunningTaskList);
        } catch (Exception e) {
            throw new BusinessException(ErrorCode.PARAMS_ERROR, "getYarnRunningTaskList failed: " + e.getMessage());
        }
    }
    /**
     * 主机接口
     */
    @GetMapping("/hosts")
    public BaseResponse<List<Map>> getHostsList(HttpServletRequest request) {

        log.info("=>调用: /getHostsList");
        List<Map> clusterHosts = new MrsClusterHosts().getHostsList();
        return ResultUtils.success(clusterHosts);
    }
}
